package com.zyy.reactivedemo.config;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author zyy
 * @date 2020/12/22 下午4:18
 * @description
 */
public class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<Integer,String> {
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {

        //保存订阅关系，需要它来给发布者响应
        this.subscription = subscription;
        //请求一个数据
        this.subscription.request(1);
    }

    @Override
    public void onNext(Integer item) {
        //接收到一个数据，处理
        System.out.println("处理接收到的数据"+item);
        //过滤掉小于0的，然后发布出去
        if(item >0 ){
            this.submit("转换后的数据:"+item);
        }

        //处理完调用request再请求一个数据
        this.subscription.request(1);
        //或者已经达到了目标，告诉cancle告诉发布者不再接收数据了
//        this.subscription.cancel();

    }

    @Override
    public void onError(Throwable throwable) {
        //出现了异常。例如处理数据的过程中出现了异常
        throwable.printStackTrace();
        //我们可以告诉发布者，后面不接受数据了
        this.subscription.cancel();
    }

    @Override
    public void onComplete() {
        //数据全部处理完了（发布者关闭了）
        System.out.println("处理器处理完了！");
        //关闭发布者
        this.close();
    }
}
